package com.flink.stream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
flink有限流
 */
public class FlinkStreamLimit {
    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //读取文件，参数是wc.txt文件的路径,readTextFile方法显示已经过时了，不过不用管
        DataStreamSource<String> line_DS = env.readTextFile("wc.txt");
        /**
         * 使用flatMap算子进行处理
         * FlatMapFunction<String, Tuple2<String, Integer>>
         * 第一个参数String表示进来的数据类型是String类型的
         * 第二个参数Tuple2<String, Integer>表示出去的类型是二元组类型的，二元组中第一个参数是String类型第二个参数是Integer类型
         * 在读取到数据之后需要转换成元祖然后再进行统计,每个单词放入元祖中都放一个1
         * 例如(hello,1) (java,1) (hello,1) (flink,1)
         * 然后分组统计后面的数字就知道每个单子出现了多少次
         */
        SingleOutputStreamOperator<Tuple2<String, Integer>> tuple2SingleOutputStreamOperator = line_DS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            /**
             * 重写flatMap方法
             * flatMap(String value, Collector<Tuple2<String, Integer>> out)
             * String value表示进来的参数value是String类型的
             * Collector<Tuple2<String, Integer>> out Collector是收集器，出去的参数out是二元组类型
             * @param value
             * @param out
             * @throws Exception
             */
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                //使用split按照空格进行拆分
                String[] words = value.split(" ");
                //拆分的结果是一个数组，然后遍历拆分的结果，也就是遍历出来每一个单词
                for (String word : words) {
                    //将单词添加到二元组中，后面写1，就到达了(hello,1) (java,1) (hello,1) (flink,1)这样的效果
                    Tuple2<String, Integer> tuple2_of = Tuple2.of(word, 1);
                    //向下游发送数据，collect方法就是把数据发送到下游
                    out.collect(tuple2_of);
                }
            }
        });

        /**
         * 跟批处理的聚合算子不一样，流处理中使用的是keyBy
         * keyBy中的参数跟flatmap一样是一个接口，是KeySelector接口，所以需要去实现这个接口
         * KeySelector<Tuple2<String, Integer>, String>
         * 第一个参数Tuple2<String, Integer>表示进来的数据是一个什么类型的，进来的数据是一个二元组
         * (hello,1) 二元组中第一个参数是String类型第二个参数是Integer类型，所以要写成Tuple2<String, Integer>
         * 第二个参数String表示要分组的参数的类型，因为要对二元组中的第一个参数进行分组也就是(hello,1)中的hello分组
         * 所以数据类型应该是String类型
         */
        KeyedStream<Tuple2<String, Integer>, String> word_by = tuple2SingleOutputStreamOperator.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            //重写getKey方法 Tuple2<String, Integer> value表示传递进来的参数是一个二元组
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                /**
                 * 返回结果，返回的结果就是要根据谁进行分组
                 * 这里根据二元组的下标就不是通过0和1了，f0就表示下标0，f1就表示下标1，f2就表示下标2
                 * 因为不需要进行其他处理，只需要通过二元组中第一个参数分组就行，所以直接value.f0
                 */
                return value.f0;
            }
        });

        //跟批处理的聚合一样，因为要对二元组中的第二个参数聚合，第二个参数的下标是1，直接写1
        SingleOutputStreamOperator<Tuple2<String, Integer>> word_sum = word_by.sum(1);
        word_sum.print();

        //启动流处理，需要抛异常
        env.execute();
    }
}